package com.yuan.kafkastudy.KafkaTest;

import com.fasterxml.jackson.databind.ser.std.StringSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Test;

import java.util.Properties;

public class PartitionProducer {
    private static final String TOPIC = "demo-partitioned";

    @Test
    public void test() {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        try (KafkaProducer<String, String> prod = new KafkaProducer<>(p)) {
            for (int i = 0; i < 10; i++) {
                String key = "key" + (i % 4);          // 4 个 key → 3 分区，会冲突可观察
                String value = "msg-" + i;
                ProducerRecord<String, String> rec =
                        new ProducerRecord<>(TOPIC, key, value);

                prod.send(rec, (meta, ex) -> {
                    if (ex == null) {
                        System.out.printf("Sent %s->%s | partition=%d offset=%d%n",
                                key, value, meta.partition(), meta.offset());
                    }
                });
            }
        }
    }
}